/*
 * Copyright (c) 2012-2015 The original author or authors
 * ------------------------------------------------------
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Apache License v2.0 which accompanies this distribution.
 *
 * The Eclipse Public License is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * The Apache License v2.0 is available at
 * http://www.opensource.org/licenses/apache2.0.php
 *
 * You may elect to redistribute this code under either of these licenses.
 */
package com.parsechina.hpush.broker.spi.persistence.h2;

import com.parsechina.hpush.broker.config.MqttConstants;
import com.parsechina.hpush.broker.internal.MessageGUID;
import com.parsechina.hpush.broker.internal.StoredMessage;
import com.parsechina.hpush.broker.internal.subscriptions.Subscription;
import com.parsechina.hpush.broker.internal.unique.ServerYeast;
import com.parsechina.hpush.broker.spi.IMessagesStore;
import com.parsechina.hpush.broker.spi.persistence.SessionsStoreUtils;
import org.h2.mvstore.MVStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentMap;

/**
 * IMessagesStore implementation backed by MapDB.
 *
 * @author linfeng
 */
class H2MessagesStore implements IMessagesStore {

    private static final Logger LOG = LoggerFactory.getLogger(H2MessagesStore.class);

    /**
     * 存储
     */
    private final MVStore mvStore;

    /**
     * topic -> MessageGUID
     */
    private ConcurrentMap<String, MessageGUID> topicMessageGuidMap;
    /**
     * maps guid to message, it's message store
     */
    private ConcurrentMap<MessageGUID, StoredMessage> messageGuidStoredMessageMap;


    H2MessagesStore(MVStore mvStore) {
        this.mvStore = mvStore;
    }

    @Override
    public void initStore() {
        messageGuidStoredMessageMap = mvStore.openMap(MqttConstants.CACHE_PERSISTED_MESSAGES_KEY);
        topicMessageGuidMap = mvStore.openMap(MqttConstants.CACHE_RETAINED_KEY);
    }

    @Override
    public void storeRetained(String topic, MessageGUID guid) {

        if (guid == null) {
            cleanRetained(topic);
        } else {
            MessageGUID oldGuid = topicMessageGuidMap.get(topic);
            if (oldGuid != null) {
                messageGuidStoredMessageMap.remove(oldGuid);
            }
            topicMessageGuidMap.put(topic, guid);
        }

    }

    @Override
    public void cleanRetained(String topic) {

        MessageGUID guid = topicMessageGuidMap.get(topic);
        if (guid != null) {
            messageGuidStoredMessageMap.remove(guid);
        }
        topicMessageGuidMap.remove(topic);
    }

    @Override
    public Collection<StoredMessage> searchMessage(Subscription condition) {

        List<StoredMessage> results = new ArrayList<>();
        String topic = condition.getTopicFilter();

        MessageGUID guid = topicMessageGuidMap.get(topic);
        if (guid != null) {
            StoredMessage storedMsg = messageGuidStoredMessageMap.get(guid);
            results.add(storedMsg);
        }

        return results;

    }

    @Override
    public MessageGUID storePublishForFuture(StoredMessage evt) {

        LOG.debug("storePublishForFuture store evt {}", evt);
        if (evt.getClientId() == null) {
            LOG.error("persisting a message without a clientId, bad programming error msg: {}", evt);
            throw new IllegalArgumentException("\"persisting a message without a clientId, bad programming error");
        }
        MessageGUID guid = new MessageGUID(ServerYeast.yeast());
        evt.setGuid(guid);
        LOG.debug("storePublishForFuture guid <{}>", guid);
        messageGuidStoredMessageMap.put(guid, evt);

        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = mvStore.openMap(SessionsStoreUtils.messageId2GuidsMapName(evt.getClientId()));
        messageIdToGuid.put(evt.getMessageID(), guid);

        return guid;
    }

    @Override
    public List<StoredMessage> listMessagesInSession(Collection<MessageGUID> guids) {
        List<StoredMessage> ret = new ArrayList<>();
        for (MessageGUID guid : guids) {
            ret.add(messageGuidStoredMessageMap.get(guid));
        }
        return ret;
    }

    @Override
    public void dropMessagesInSession(String clientID) {
        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = mvStore.openMap(SessionsStoreUtils.messageId2GuidsMapName(clientID));
        if (messageIdToGuid != null) {
            for (MessageGUID guid : messageIdToGuid.values()) {
                removeStoredMessage(guid);
            }
            messageIdToGuid.clear();
        }
    }

    @Override
    public void removeStoredMessage(MessageGUID guid) {
        //remove only the not retained and no more referenced
        StoredMessage storedMessage = messageGuidStoredMessageMap.get(guid);
        if (storedMessage != null && !storedMessage.isRetained() && storedMessage.getReferenceCounter() <= 0) {
            if (storedMessage.getReferenceCounter() < 0) {
                LOG.error("we should never have gotten a reference count less than zero");
            }
            LOG.debug("Cleaning not retained message guid {}", guid);
            messageGuidStoredMessageMap.remove(guid);
        }
    }

    @Override
    public StoredMessage getMessageByGuid(MessageGUID guid) {
        return messageGuidStoredMessageMap.get(guid);
    }



    @Override
    public void incUsageCounter(MessageGUID guid) {
        StoredMessage storedMessage = messageGuidStoredMessageMap.get(guid);
        storedMessage.incReferenceCounter();
        messageGuidStoredMessageMap.put(guid, storedMessage);
    }

    @Override
    public void decUsageCounter(MessageGUID guid) {
        StoredMessage storedMessage = messageGuidStoredMessageMap.get(guid);
        storedMessage.decReferenceCounter();
        messageGuidStoredMessageMap.put(guid, storedMessage);
    }
}
